package com.bigdata.sql

import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo16Reason {

  def main(args: Array[String]): Unit = {

    // 使用sparkSQL，需要创建一个SparkSession的对象，里面包含了SparkContext对象
    // SparkSQL底层还是由RDD来执行的，RDD之间会存在Shuffle，SparkSQL默认的shuffle分区数是200
    // 这个200的值我们可以使用spark.sql.shuffle.partitions参数调整
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("wc-sql")
      //指定spark sql  在shuffle之后分区数，默认时200，类似reduce的数量
      .config("spark.sql.shuffle.partitions", 3)
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("error")

    // SparkSQL支持非常丰富的数据源，这里使用csv文件（一种文本文件）
    val linesDF: DataFrame = spark
      .read
      .format("csv")
      .schema("lines string")
      .option("sep", "-") // 一行数据做为一列
      .load("data/reason.txt")

    // 查看表的结构
    linesDF.printSchema()
    // 默认查看20行，如果一行数据太多了，显示不全
//    linesDF.show(10000, false)

    // 写SQL实现wordcount, view是逻辑上的，本身是没有持久化的
    linesDF.createOrReplaceTempView("reason")

    val resultDF: DataFrame = spark.sql(
      """
        |select lines,count(*) number from reason group by lines order by number desc
        |""".stripMargin)

    resultDF.show(10000, false)

    // 200, sparksql 如果有shuffle，那么分区数默认是200
    println(resultDF.rdd.getNumPartitions)

//    resultDF
//      .write
//      .format("csv")
//      .option("sep", "-")
//      .mode(SaveMode.Overwrite)
//      .save("data/sql")


  }

}
